#!/usr/bin/env python
# -*- coding: utf-8 -*-
# @Author  : HaiFeng
# @Email   : 24918700@qq.com
# @Time    : 2019/7/3
# @desc    : data server by zmq


import zmq
import gzip
from pandas import DataFrame
import time
import json
from config import log
import config as cfg
from redis import Redis


class Server(object):
    """"""

    def __init__(self):
        context = zmq.Context(1)
        self.server = context.socket(zmq.REP)
        self.server.bind(f'tcp://*:{cfg.zmq_port}')
        self.days = [d[0] for d in cfg.mysql_fetchall(cfg.sql_trade_date)]

    def run(self):
        log.war('listen to port: {}'.format(self.server.LAST_ENDPOINT.decode()))
        while True:
            request = self.server.recv_json()  # .recv_string()
            log.info(request)
            # Min, Day, Real, Time, Product, TradeDate, InstrumentInfo, Instrumet888, Rate000
            rsp = self.read_data(request)
            if rsp == '':
                log.error('{} 未取得数据'.format(request))
            rsp = gzip.compress(rsp.encode(), 9)
            self.server.send(rsp)

    def read_data(self, req) -> str:
        """ Min, Day, Real, Time, Product, TradeDate, InstrumentInfo, Instrumet888, Rate000 """
        params = None
        if req['Type'] in [0, 1]:  # Min, Day
            # tradingday 为 varchar 需要用in 不能用between
            ds = [d for d in self.days if d >= req['Begin'] and d <= req['End']]
            ds = "','".join(ds)
            sql = cfg.sql_min.format(*[req['Instrument'], f"'{ds}'"])
        elif req['Type'] == 2:  # Real
            v = cfg.rds.lrange(req['Instrument'], 0, -1)
            # 兼容旧版
            return ('[' + ','.join(v) + ']').replace('TradingDay', 'Tradingday')  # 转换为str
        elif req['Type'] == 3:  # Time
            sql = cfg.sql_time
        elif req['Type'] == 4:  # Product==>instrument
            sql = cfg.sql_product
        elif req['Type'] == 5:  # TradeDate
            sql = cfg.sql_trade_date
        elif req['Type'] == 6:  # InstrumentInfo
            sql = cfg.sql_inst_proc
        elif req['Type'] == 7:  # Instrumet888
            sql = cfg.sql_888
        elif req['Type'] == 8:  # rate000
            sql = cfg.sql_rate
        else:
            return ''
        # 调用前需重复调用Create engine 否则报错: Operation cannot be accomplished in current state
        df: DataFrame = cfg.mysql_df(sql)

        # K线
        if req['Type'] <= 2:
            # 20181010 采用yyyy-mm-dd格式,无需转换.df['_id'] = df['_id'].apply(lambda x: ''.format(x[0:4] + x[5:7] + x[8:]))  # ==> yyyyMMdd HH:mm:ss
            if req['Type'] == 1:  # 日线
                df_tmp: DataFrame = DataFrame()
                g = df.groupby(by=df['TradingDay'])
                # df_tmp.index = g.indices
                df_tmp['_id'] = g.indices
                df_tmp['Open'] = g['Open'].first()
                df_tmp['High'] = g['High'].max()
                df_tmp['Low'] = g['Low'].min()
                df_tmp['Close'] = g['Close'].last()
                df_tmp['Volume'] = g['Volume'].sum()
                df_tmp['OpenInterest'] = g['OpenInterest'].last()
                df_tmp = df_tmp.reindex(df_tmp['_id'])
                df = df_tmp

            keys = ["_id", "Instrument", "TradingDay", "High", "Low", "Open", "Close", "Volume", "OpenInterest"]
            df = df.loc[:, keys]
        rtn = ''
        if req['Type'] == 5:
            rtn = [r for r in df['day']]
            rtn = json.dumps(rtn)
        elif req['Type'] == 8:  # => dict
            rtn = df.to_json(orient='values')
            rtn = '{{{}}}'.format(rtn.replace('[', '').replace(']', '').replace('",', '":'))
        elif req['Type'] == 3:
            # df['WorkingTimes'] = df['WorkingTimes'].apply(str)
            df = df[['GroupId', 'WorkingTimes']]
            rtn = df.to_json(orient='records')
            rtn = rtn.replace('\\"', '"').replace('"[', '[').replace(']"', ']')
        elif req['Type'] == 4:
            df['VolumeTuple'] = df['VolumeTuple'].apply(int)
            df = df[['_id', 'PriceTick', 'VolumeTuple', 'ExchangeID', 'ProductType']]
            rtn = df.to_json(orient='records')
        else:  # 4
            rtn = df.to_json(orient='records')
        return rtn


if __name__ == '__main__':
    Server().run()
